package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.Constant;
import com.heima.common.constants.KafkaConstant;
import com.heima.common.constants.RedisConstant;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.ITaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class TaskServiceIpml implements ITaskService {

    @Resource
    private TaskinfoLogsMapper taskinfoLogsMapper;
    @Resource
    private TaskinfoMapper taskinfoMapper;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private KafkaTemplate kafkaTemplate;

    private final String FREFIX = "start";

    /**
     * 延时任务
     * @param task
     * @return
     */
    @Override
    @Transactional
    public Long TaskAdd(Task task) {
        //防止多次点击，幂等性方案一
        //可用redis分布锁
        //新增两张表
        //
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task,taskinfo);
        taskinfoMapper.insert(taskinfo);

        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo,taskinfoLogs);
        taskinfoLogs.setStatus(TaskinfoLogs.SCHEDULED);
        taskinfoLogsMapper.insert(taskinfoLogs);
        //

        long time = LocalDateTime.now().plusMinutes(Constant.NUMBER_FIVE)
                .toInstant(ZoneOffset.of(Constant.TIME_UP_ENGIHT)).toEpochMilli();
        if (task.getExecuteTime().getTime() <= System.currentTimeMillis()){
            //消息队列
            log.info("时间已过，mq通知开启");
            sendTask(JSON.toJSONString(taskinfo));
        }else if (task.getExecuteTime().getTime() <= time){
            log.info("时间将近，进入Redis开赛");
            stringRedisTemplate.opsForZSet()
                    .add(FREFIX+task.getTaskType(), JSON.toJSONString(taskinfo),task.getExecuteTime().getTime());
        }

        return taskinfo.getTaskId();
    }

    /**
     * mq消息通知(延时任务开始执行)
     * @param task
     */
    @Transactional
    public void sendTask(String task) {
        //删除taskinfo表，修改log表状态
        //再通知发布
        Taskinfo taskinfo = JSON.parseObject(task, Taskinfo.class);
        taskinfoMapper.deleteById(taskinfo.getTaskId());

        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        taskinfoLogs.setTaskId(taskinfo.getTaskId());
        taskinfoLogs.setStatus(TaskinfoLogs.EXECUTED);
        taskinfoLogsMapper.updateById(taskinfoLogs);
        //
        kafkaTemplate.send(KafkaConstant.TOPIC_ARTICLE_UP_OK,taskinfo.getParameters());
    }

    @Scheduled(cron = "0/30 * * * * ?")
    public void autoTaskToRedis(){
        log.info("开启定时任务");
        //redis分布锁（状态）
        Boolean redis = stringRedisTemplate.opsForValue().setIfAbsent
                (RedisConstant.SYS_AUTO_TASK_REDIS, RedisConstant.SYS_REDIS_VALUE, RedisConstant.SYS_REDIS_DISTIME, TimeUnit.MINUTES);
        if (null != redis && !redis){
            log.info("你没锁，真可怜");
            return;
        }

        try {
            LambdaQueryWrapper<Taskinfo> lambdaQuery = Wrappers.<Taskinfo>lambdaQuery();
            List<Taskinfo> taskinfos = taskinfoMapper.selectList(lambdaQuery);
            if (CollectionUtils.isEmpty(taskinfos)){
                log.info("任务列表为空，不执行操作");
                return;
            }
            long time = LocalDateTime.now().plusMinutes(Constant.NUMBER_FIVE)
                    .toInstant(ZoneOffset.of(Constant.TIME_UP_ENGIHT)).toEpochMilli();
            for (Taskinfo taskinfo : taskinfos) {
                //可以在查询数据库的时候就做了
                if (taskinfo.getExecuteTime().getTime() <= time){
                    stringRedisTemplate.opsForZSet()
                            .add(FREFIX+taskinfo.getTaskType(), JSON.toJSONString(taskinfo),taskinfo.getExecuteTime().getTime());
                }
            }
        } finally {
            //解锁一定要执行，则需要try{}fanily{}来执行
            stringRedisTemplate.delete(RedisConstant.SYS_AUTO_TASK_REDIS);
        }
        log.info("进入Redis开赛");
    }

    @Scheduled(cron = "0/30 * * * * ?")
    public void autoRedisCheck(){
        Boolean check = stringRedisTemplate.opsForValue().setIfAbsent
                (RedisConstant.SYS_AUTO_REDIS_CHECK, RedisConstant.SYS_REDIS_VALUE, RedisConstant.SYS_REDIS_DISTIME, TimeUnit.MINUTES);
        if (null != check && !check){
            log.info("没抢到锁，悲报");
            return;
        }

        try {
            //获取key
            Set<String> keys = scan(FREFIX);
            if (CollectionUtils.isEmpty(keys)){
                return;
            }
            for (String key : keys) {
                //
                Set<String> strings = stringRedisTemplate
                        .opsForZSet().rangeByScore(key, Constant.NUMBER_ZERO, System.currentTimeMillis());
                if (CollectionUtils.isEmpty(strings)){
                    log.info("gaikey的时间还未到，{}",key);
                    continue;
                }
                //
                for (String string : strings) {
                    try {
                        sendTask(string);
                        stringRedisTemplate.opsForZSet().remove(key,string);
                    } catch (Exception e) {
                        log.info("mq消息通知异常或者删除redis里的键值对失败");
                    }
                }
            }
        } finally {
            stringRedisTemplate.delete(RedisConstant.SYS_AUTO_REDIS_CHECK);
        }

    }

    private Set<String> scan(String patten) {
        Set<String> keys = stringRedisTemplate.execute((RedisCallback<Set<String>>) connection -> {
            Set<String> result = new HashSet<>();
            try (Cursor<byte[]> cursor = connection.scan(new ScanOptions.ScanOptionsBuilder()
                    .match(patten).count(Constant.NUMBER_TEN_THOUSAND).build())) {
                while (cursor.hasNext()) {
                    result.add(new String(cursor.next()));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return result;
        });
        return keys;
    }
}
